#!/bin/bash

# Kafka主题创建后的Maxwell验证脚本

echo "=== Kafka主题创建后的Maxwell验证 ==="
echo "时间: $(date)"
echo ""

# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color

# 配置变量
MYSQL_HOST="192.168.10.102"
MYSQL_PORT="3306"
MYSQL_USER="root"
MYSQL_PASSWORD="000000"
MAXWELL_USER="maxwell"
MAXWELL_PASSWORD="maxwell"
DATABASE="smart_tour"
KAFKA_TOPIC="maxwell_smart_tour"
MAXWELL_HOME="/opt/module/maxwell"

echo -e "${GREEN}✓ Kafka主题 'maxwell_smart_tour' 已创建成功${NC}"
echo ""

# 1. 验证Kafka主题
echo -e "${BLUE}1. 验证Kafka主题${NC}"
echo "----------------------------------------"

# 检查主题详情
echo "检查主题详情..."
kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic $KAFKA_TOPIC

# 检查主题列表
echo "检查主题列表..."
kafka-topics.sh --bootstrap-server hadoop102:9092 --list | grep $KAFKA_TOPIC

echo ""

# 2. 检查Maxwell配置
echo -e "${BLUE}2. 检查Maxwell配置${NC}"
echo "----------------------------------------"

if [ -f "$MAXWELL_HOME/config.properties" ]; then
    echo "配置文件存在，检查关键配置..."
    echo "----------------------------------------"
    grep -E "^(host|port|user|password|filter|kafka_topic)=" "$MAXWELL_HOME/config.properties" || echo "未找到关键配置项"
    echo "----------------------------------------"
    
    # 检查过滤规则
    if grep -q "filter=include:smart_tour" "$MAXWELL_HOME/config.properties"; then
        echo -e "${GREEN}✓ 过滤规则配置正确${NC}"
    else
        echo -e "${RED}✗ 过滤规则配置错误或缺失${NC}"
        echo "请确保配置文件中包含: filter=include:smart_tour.*"
    fi
    
    # 检查Kafka主题配置
    if grep -q "kafka_topic=$KAFKA_TOPIC" "$MAXWELL_HOME/config.properties"; then
        echo -e "${GREEN}✓ Kafka主题配置正确${NC}"
    else
        echo -e "${YELLOW}⚠ Kafka主题配置可能不正确${NC}"
        echo "请确保配置文件中包含: kafka_topic=$KAFKA_TOPIC"
    fi
else
    echo -e "${RED}✗ 配置文件不存在: $MAXWELL_HOME/config.properties${NC}"
fi

echo ""

# 3. 检查Maxwell进程
echo -e "${BLUE}3. 检查Maxwell进程${NC}"
echo "----------------------------------------"

PIDS=$(pgrep -f maxwell)
if [ -n "$PIDS" ]; then
    echo -e "${GREEN}✓ Maxwell进程正在运行 (PID: $PIDS)${NC}"
    ps aux | grep maxwell | grep -v grep
else
    echo -e "${YELLOW}⚠ Maxwell进程未运行${NC}"
    echo "启动Maxwell..."
    
    if [ -f "$MAXWELL_HOME/config.properties" ]; then
        # 创建日志目录
        mkdir -p "$MAXWELL_HOME/logs"
        
        # 启动Maxwell
        nohup $MAXWELL_HOME/bin/maxwell --config="$MAXWELL_HOME/config.properties" > "$MAXWELL_HOME/logs/maxwell.log" 2>&1 &
        MAXWELL_PID=$!
        
        sleep 3
        
        if kill -0 $MAXWELL_PID 2>/dev/null; then
            echo -e "${GREEN}✓ Maxwell启动成功 (PID: $MAXWELL_PID)${NC}"
        else
            echo -e "${RED}✗ Maxwell启动失败${NC}"
            echo "请检查日志: $MAXWELL_HOME/logs/maxwell.log"
        fi
    else
        echo -e "${RED}✗ 配置文件不存在，无法启动Maxwell${NC}"
    fi
fi

echo ""

# 4. 检查Maxwell日志
echo -e "${BLUE}4. 检查Maxwell日志${NC}"
echo "----------------------------------------"

if [ -f "$MAXWELL_HOME/logs/maxwell.log" ]; then
    echo "Maxwell日志最后10行:"
    echo "----------------------------------------"
    tail -10 "$MAXWELL_HOME/logs/maxwell.log"
    echo "----------------------------------------"
    
    # 检查错误
    if grep -i error "$MAXWELL_HOME/logs/maxwell.log" | tail -1; then
        echo -e "${RED}⚠ 发现错误日志${NC}"
    else
        echo -e "${GREEN}✓ 日志无错误${NC}"
    fi
else
    echo -e "${YELLOW}⚠ Maxwell日志文件不存在${NC}"
fi

echo ""

# 5. 测试数据库连接
echo -e "${BLUE}5. 测试数据库连接${NC}"
echo "----------------------------------------"

# 测试maxwell用户连接
if mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MAXWELL_USER -p$MAXWELL_PASSWORD -e "SELECT 1;" 2>/dev/null; then
    echo -e "${GREEN}✓ Maxwell用户连接正常${NC}"
else
    echo -e "${RED}✗ Maxwell用户连接失败${NC}"
fi

# 测试smart_tour数据库访问
if mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MAXWELL_USER -p$MAXWELL_PASSWORD -e "USE $DATABASE; SELECT 1;" 2>/dev/null; then
    echo -e "${GREEN}✓ Smart_tour数据库可访问${NC}"
else
    echo -e "${RED}✗ Smart_tour数据库不可访问${NC}"
fi

echo ""

# 6. 创建测试环境
echo -e "${BLUE}6. 创建测试环境${NC}"
echo "----------------------------------------"

echo "创建测试表..."
mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASSWORD << EOF
USE $DATABASE;
CREATE TABLE IF NOT EXISTS maxwell_test (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(100),
    value TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
EOF

if [ $? -eq 0 ]; then
    echo -e "${GREEN}✓ 测试表创建成功${NC}"
else
    echo -e "${RED}✗ 测试表创建失败${NC}"
fi

echo ""

# 7. 启动Kafka消费者监控
echo -e "${BLUE}7. 启动Kafka消费者监控${NC}"
echo "----------------------------------------"

echo "启动Kafka消费者监控..."
echo "监控主题: $KAFKA_TOPIC"
echo "按 Ctrl+C 停止监控"
echo "----------------------------------------"

# 启动消费者（后台运行）
kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic $KAFKA_TOPIC --from-beginning > /tmp/maxwell_output.log 2>&1 &
CONSUMER_PID=$!

echo "Kafka消费者已启动 (PID: $CONSUMER_PID)"
echo "输出文件: /tmp/maxwell_output.log"

# 等待消费者启动
sleep 3

echo ""

# 8. 执行测试操作
echo -e "${BLUE}8. 执行测试操作${NC}"
echo "----------------------------------------"

echo "等待5秒让Maxwell准备就绪..."
sleep 5

echo "执行数据库操作..."

# INSERT操作
echo "1. 执行INSERT操作..."
mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASSWORD << EOF
USE $DATABASE;
INSERT INTO maxwell_test (name, value) VALUES ('Maxwell测试', '这是Maxwell验证测试数据');
EOF

sleep 2

# UPDATE操作
echo "2. 执行UPDATE操作..."
mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASSWORD << EOF
USE $DATABASE;
UPDATE maxwell_test SET value = '这是更新后的Maxwell测试数据' WHERE name = 'Maxwell测试';
EOF

sleep 2

# DELETE操作
echo "3. 执行DELETE操作..."
mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASSWORD << EOF
USE $DATABASE;
DELETE FROM maxwell_test WHERE name = 'Maxwell测试';
EOF

echo -e "${GREEN}✓ 测试操作执行完成${NC}"

echo ""

# 9. 检查输出结果
echo -e "${BLUE}9. 检查输出结果${NC}"
echo "----------------------------------------"

echo "等待5秒让Maxwell处理完成..."
sleep 5

# 停止消费者
kill $CONSUMER_PID 2>/dev/null

# 检查输出
if [ -f "/tmp/maxwell_output.log" ]; then
    echo "Kafka输出内容:"
    echo "----------------------------------------"
    cat /tmp/maxwell_output.log
    echo "----------------------------------------"
    
    # 分析输出
    echo "分析输出结果..."
    
    if grep -q "INSERT" /tmp/maxwell_output.log; then
        echo -e "${GREEN}✓ 检测到INSERT操作${NC}"
    else
        echo -e "${RED}✗ 未检测到INSERT操作${NC}"
    fi
    
    if grep -q "UPDATE" /tmp/maxwell_output.log; then
        echo -e "${GREEN}✓ 检测到UPDATE操作${NC}"
    else
        echo -e "${RED}✗ 未检测到UPDATE操作${NC}"
    fi
    
    if grep -q "DELETE" /tmp/maxwell_output.log; then
        echo -e "${GREEN}✓ 检测到DELETE操作${NC}"
    else
        echo -e "${RED}✗ 未检测到DELETE操作${NC}"
    fi
    
    if grep -q "smart_tour" /tmp/maxwell_output.log; then
        echo -e "${GREEN}✓ 检测到smart_tour数据库操作${NC}"
    else
        echo -e "${RED}✗ 未检测到smart_tour数据库操作${NC}"
    fi
    
    # 统计输出数量
    OUTPUT_COUNT=$(wc -l < /tmp/maxwell_output.log)
    echo "输出记录数: $OUTPUT_COUNT"
    
else
    echo -e "${YELLOW}⚠ 未找到Kafka输出文件${NC}"
fi

echo ""

# 10. 验证总结
echo -e "${BLUE}10. 验证总结${NC}"
echo "----------------------------------------"

echo "验证项目:"
echo "  ✓ Kafka主题创建"
echo "  ✓ Maxwell配置检查"
echo "  ✓ Maxwell进程检查"
echo "  ✓ 数据库连接测试"
echo "  ✓ 测试环境创建"
echo "  ✓ 数据库操作执行"
echo "  ✓ 输出结果分析"

echo ""
echo "如果所有测试都通过，说明Maxwell配置正确并能正常捕捉smart_tour数据库的变化。"
echo ""
echo "后续操作建议:"
echo "1. 监控Maxwell日志: tail -f $MAXWELL_HOME/logs/maxwell.log"
echo "2. 监控Kafka主题: kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic $KAFKA_TOPIC --from-beginning"
echo "3. 检查Maxwell状态: ps aux | grep maxwell"
echo "4. 查看数据库变化: mysql -h $MYSQL_HOST -P $MYSQL_PORT -u $MYSQL_USER -p$MYSQL_PASSWORD -e 'USE $DATABASE; SELECT * FROM maxwell_test;'"

echo ""
echo "=== 验证完成 ===" 